package monitor

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	_ "embed"
	"encoding/json"
	"fmt"
	. "gitee.com/aldrice/monitor/configs"
	"gitee.com/aldrice/monitor/ent"
	"gitee.com/aldrice/monitor/ent/metric"
	"gitee.com/aldrice/monitor/ent/object"
	"gitee.com/aldrice/monitor/ent/setting"
	. "gitee.com/aldrice/monitor/internal/common"
	. "gitee.com/aldrice/monitor/internal/monitor/definition"
	. "gitee.com/aldrice/monitor/internal/monitor/report"
	"gitee.com/aldrice/monitor/internal/registry"
	"gitee.com/aldrice/monitor/pkg/logging"
	"github.com/dgrijalva/jwt-go"
	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"io/ioutil"
	"net"
	"strconv"
	"time"
)

const storeTimeout = 500 // unit: ms

const (
	keyPubAddress = "pub_ip"
	keyLocAddress = "loc_ip"
	machineId     = "mid"
)

type server struct {
	requestMap map[string]*ReportReq
}

// Report 接收模块
func (s *server) Report(ctx context.Context, req *ReportReq) (*ReportRsp, error) {
	fmt.Println("上报数据：", req)

	// 将handler整合进来
	// 用私钥解析token, 若无法正确解析则抛掉这则报告, 否则尝试获取对应的IP地址
	claim := &jwt.StandardClaims{}
	token, err := jwt.ParseWithClaims(req.GetToken(), claim, JwtObjectKey)
	if err != nil || !token.Valid {
		return responseFormat(CodeTransmitDenied, "refuse to store unidentified report", nil), err
	}
	db := registry.Ent()
	uid, err := strconv.Atoi(claim.Audience)
	obj, err := db.Object.Query().Where(object.IDEQ(uid)).First(ctx)
	if err != nil {
		return responseFormat(CodeTransmitDenied, "failed to acquired the entity of object", nil), err
	}

	ms, err := db.Metric.Query().Where(
		metric.HasAggregatedByWith(
			setting.EnableEQ(true),
			setting.HasWithObjectWith(object.IDEQ(obj.ID)),
		),
	).All(ctx)
	if err != nil {
		return responseFormat(CodeTransmitDenied, "failed to acquired the entity of metrics", nil), err
	}

	sts := make([]string, len(ms))
	for i, item := range ms {
		sts[i] = item.Name
	}

	st := &AgentSetting{
		ReportInterval: int32(obj.ReportInterval),
		ResendTimes:    int32(obj.ResendTimes),
		FetchTimeout:   int32(obj.FetchTimeout),
		Settings:       sts,
	}

	// 更新当前object的ip信息
	_, err = obj.Update().
		SetPublicAddress(req.Dimensions[keyPubAddress]).
		SetLocalAddress(req.Dimensions[keyLocAddress]).
		SetLastUpdatedAt(time.Now()).
		Save(ctx)
	if err != nil {
		return responseFormat(CodeFailedToHandle, "failed to update object address", st), err
	}

	// 若传来的报文没有爬取数据的值, 则直接返回
	if len(req.GetRequests()) == 0 {
		return responseFormat(CodeSuccess, "", st), nil
	}

	timestamp := req.Timestamp
	// 将上报数据持久化到influxdb
	for _, item := range req.GetRequests() {
		// store request
		if err := storeRequest(uid, item, timestamp, req.Dimensions); err != nil {
			return responseFormat(CodeFailedToStored, "failed to store agent's operating data", st), err
		}
	}

	// 将obj的id及上报时间戳放入dataProcess队列中
	dataProcessMQ := registry.GetDataProcessMQ()
	marshal, err := json.Marshal(ProcessMsg{Id: uid, Timestamp: timestamp})
	if err != nil {
		return responseFormat(CodeFailedToHandle, "failed to push the data into message queue", st), err
	}
	if err := dataProcessMQ.PublishSimple(marshal); err != nil {
		fmt.Println(err)
		return responseFormat(CodeFailedToHandle, "failed to put obj id into MQ", st), err
	}
	// 返回处理结果给agent
	return responseFormat(CodeSuccess, "", st), nil
}

// storeRequest write report item into influxdb
func storeRequest(uid int, item *ReportReqItem, ts int64, mp map[string]string) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*storeTimeout)
	defer cancel()

	p := influxdb2.NewPointWithMeasurement(Config.InfluxDB.Bucket).
		AddField(item.GetMetric(), item.GetValue()).
		SetTime(time.Unix(ts, 0))

	// add all tags that item has inside the point
	for key, val := range mp {
		p.AddTag(key, val)
	}
	p.AddTag(machineId, string(rune(uid)))

	return (*registry.GetWriteAPI()).WritePoint(ctx, p)
}

func InitReceiver(isSafeConn bool) (*grpc.Server, error) {
	if err := initMetrics(); err != nil {
		return nil, err
	}

	port := Config.Listen.Rpc
	listener, err := net.Listen("tcp", port)
	if err != nil {
		return nil, err
	}

	var s *grpc.Server
	if isSafeConn {
		cert, err := tls.LoadX509KeyPair(Config.TLS.CertFile, Config.TLS.KeyFile)
		if err != nil {
			return nil, fmt.Errorf("failed to load certification file, err: %s", err.Error())
		}
		certPool := x509.NewCertPool()
		ca, err := ioutil.ReadFile(Config.TLS.CaFile)
		if err != nil {
			return nil, fmt.Errorf("failed to load ca file, err: %s", err.Error())
		}
		ok := certPool.AppendCertsFromPEM(ca)
		if !ok {
			return nil, fmt.Errorf("failed to append certicication into pool")
		}

		creeds := credentials.NewTLS(&tls.Config{
			Certificates: []tls.Certificate{cert},
			ClientAuth:   tls.RequireAndVerifyClientCert,
			ClientCAs:    certPool,
		})
		s = grpc.NewServer(grpc.Creds(creeds))
	} else {
		s = grpc.NewServer()
	}

	RegisterReportServerServer(s, &server{})
	logging.Logger().Info("start gRPC service on port " + port)

	go func() {
		if err := s.Serve(listener); err != nil {
			logging.Logger().Fatal("failed to start grpc service", zap.Error(err))
		}
	}()

	return s, nil
}

func initMetrics() error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()

	var mCreates []*ent.MetricUpsertOne

	addMetricFunc := func(name string) {
		mCreates = append(mCreates,
			registry.Ent().Metric.Create().SetName(name).OnConflictColumns(metric.FieldName).Ignore())
	}
	for _, m := range Metrics {
		addMetricFunc(m)
	}

	for _, mCreate := range mCreates {
		if err := mCreate.Exec(ctx); err != nil {
			return err
		}
	}

	return nil
}
